package com.ybkj.o2o.zhuanche.conf;

import com.ybkj.o2o.zhuanche.thrift.ThriftService;
import org.apache.helix.manager.zk.ByteArraySerializer;
import org.apache.helix.manager.zk.ZkClient;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.codehaus.jackson.map.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

/**
 * Created by shine on 2016/12/5.
 */
//@Configuration
public class ZookeeperConfiguration {

    private static final Logger logger = LoggerFactory.getLogger(ZookeeperConfiguration.class);

    @Value("${zookeeper.service.name}")
    String serviceName;

    @Value("${zookeeper.server.list}")
    String zookeeperList;

    // thrift实例列表
    public static List<com.ybkj.o2o.zhuanche.conf.ThriftServiceClient> ServiceClients = new LinkedList<>();

    /**
     * 初始化.
     */
    @PostConstruct
    public void init() throws Exception {

        startZookeeper();

    }

    private void startZookeeper() throws Exception {


        boolean flag = true;
        // 会出现连接不上的情况，这里循环连接
        while (flag) {

            try {
                ZkClient zkClient = new ZkClient(zookeeperList);

                zkClient.setZkSerializer(new ByteArraySerializer());

                if (zkClient.exists(serviceName)) {

                    flag = false;

                    List<String> workers = zkClient.getChildren(serviceName);
                    //构造服务地址
                    buildServerAddress(zkClient);
                    //订阅事件变化
                    subscribeChildChanges(zkClient, workers);

                } else {
                    flag = true;
                    try {
                        zkClient.close();
                    } catch (Exception e) {
                        logger.error("close zkClient error");
                    }
                }
            } catch (Exception e) {
                logger.error("zk error", e);
            }

        }


    }

    private void buildServerAddress(ZkClient zkClient) {

        ObjectMapper mapper = new ObjectMapper();

        try {

            List<String> workers = zkClient.getChildren(serviceName);

            //订阅事件变化
            subscribeChildChanges(zkClient, workers);
            //删除所有
            if (null == workers) {
                closeClient(ServiceClients);
                return;
            }

            List<com.ybkj.o2o.zhuanche.conf.ThriftServiceClient> allServers = new LinkedList<>();

            for (String workId : workers) {
                String workIdPath = String.format("%s/%s", serviceName, workId);

                List<String> nodes = zkClient.getChildren(workIdPath);

                if (!CollectionUtils.isEmpty(nodes)) {
                    for (String node : nodes) {
                        String nodePath = String.format("%s/%s", workIdPath, node);
                        byte[] bytes = zkClient.readData(nodePath);

                        String server = new String(bytes, "utf-8");

                        com.ybkj.o2o.zhuanche.conf.SnowflakePeer peer = mapper.readValue(server, com.ybkj.o2o.zhuanche.conf.SnowflakePeer.class);

                        List<String> thriftServers = peer.getThrift();

                        if (!CollectionUtils.isEmpty(thriftServers)) {
                            //创建服务信息
                            for (String thriftServer : thriftServers) {

                                com.ybkj.o2o.zhuanche.conf.ThriftServiceClient client = new com.ybkj.o2o.zhuanche.conf.ThriftServiceClient();
                                client.setWorkerId(workId);
                                client.setThriftServer(thriftServer);
                                //添加所有的服务信息
                                allServers.add(client);

                                if (!ServiceClients.contains(client)) {
                                    client.setClient(createThriftService(thriftServer));
                                    //如果没有
                                    ServiceClients.add(client);
                                }
                            }
                        }
                    }
                }
            }

            //判断服务还存在不
            List<com.ybkj.o2o.zhuanche.conf.ThriftServiceClient> removeable = new ArrayList<>();
            //添加删除元素
            for (com.ybkj.o2o.zhuanche.conf.ThriftServiceClient client : ServiceClients) {
                if (!allServers.contains(client)) {
                    removeable.add(client);
                }
            }

            //关闭连接并删除
            closeClient(removeable);

        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    /**
     * 订阅节点变化.
     */
    private void subscribeChildChanges(ZkClient zkClient, List<String> workers) {

        zkClient.unsubscribeAll();

        zkClient.subscribeChildChanges(serviceName, (String ss, List<String> childs) -> {
            //构造服务地址
            logger.info("Snowflake Service Changed.");

            buildServerAddress(zkClient);

        });

        if (null != workers) {
            //监听每个
            for (String worker : workers) {
                zkClient.subscribeChildChanges(serviceName + "/" + worker, (String ss, List<String> childs)
                        -> {

                    logger.info("Snowflake Service Changed.");

                    buildServerAddress(zkClient);

                });
            }
        }

    }

    private void closeClient(List<com.ybkj.o2o.zhuanche.conf.ThriftServiceClient> removeable) {

        for (com.ybkj.o2o.zhuanche.conf.ThriftServiceClient client : removeable) {

            ThriftService.Client cli = client.getClient();
            try {
                cli.getInputProtocol().getTransport().close();
                cli.getOutputProtocol().getTransport().close();
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }

        //删除不存在的服务
        ServiceClients.removeAll(removeable);
    }

    // 创建一个服务实例
    private ThriftService.Client createThriftService(String serviceInstanceName) {

        String ip = serviceInstanceName.split(":")[0];
        int port = Integer.parseInt(serviceInstanceName.split(":")[1]);

        logger.info("create tsocket at {} port {}", ip, port);

        TTransport transport = null;
        try {
            transport = new TFramedTransport(new TSocket(ip, port));
            transport.open();
        } catch (TTransportException ex) {
            ex.printStackTrace();
        }

        return new ThriftService.Client(new TBinaryProtocol(transport));
    }

}
